package com.lagou.mall.esflow.server.handler;

import com.lagou.mall.esflow.server.consumer.IIndexConsumer;
import java.time.Duration;
import java.util.concurrent.ThreadPoolExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecords;

/**
 * @Author xiazhenyu
 * @Description topic 消费者处理请求工具
 * @Date 5:15 下午 2021/2/24
 **/
@Slf4j
public class TopicConsumerProcess {


    public void topicProcess(IIndexConsumer indexConsumer) {
        AbstractTopicConsumerHandler topicConsumerHandler = indexConsumer.getTopicConsumerHandler();
        ThreadPoolExecutor consumerThreadPool = indexConsumer.getThreadPoolExecutor();
        while (true) {
            try {
                //consumer 销毁，while退出
//                if (!indexConsumer.getKafkaConsumer()) {
//                    continue;
//                }
                ConsumerRecords<String, String> consumerRecords = indexConsumer.getKafkaConsumer().poll(Duration.ofMillis(1000));
                // 无消息
                if (consumerRecords.count() == 0) {
                    indexConsumer.sleep();
                    continue;
                }
                log.info("consumer  name:{},receive message size:{},poolRemainingSize:{}",
                        indexConsumer.getClass().getCanonicalName(),
                        consumerRecords.count(), consumerThreadPool.getQueue().remainingCapacity());
                consumerRecords.forEach(message -> {
                    MessageRunnable messageRunnable = new MessageRunnable(indexConsumer.getKafkaConsumer(),
                            message,
                            topicConsumerHandler);
                    // 线程池需要使用拒绝策略为 CallerRunsPolicy
                    consumerThreadPool.execute(messageRunnable);
                });
            } catch (Exception e) {
                log.error("topicProcess error", e);
            }
        }
    }
}
